package spring.cloud.tasks.executor.listener.failover;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.executor.listener.ListenerHolder;
import spring.cloud.tasks.executor.listener.config.TaskConfigurationService;
import spring.cloud.tasks.executor.node.TaskNodePaths;
import spring.cloud.tasks.executor.task.ExecuteNode;
import spring.cloud.tasks.executor.task.ExecuteService;
import spring.cloud.tasks.executor.task.TaskExecutor;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

/**
 * 失效转移监听管理器.
 */
@Slf4j
@Setter
@Getter
public class FailoverListenerHolder extends ListenerHolder {
    public final TaskConfigurationService configurationService;
    public final ExecuteService executeService;
    public final FailoverService failoverService;
    public final String executionPath;
    public final Set<String> runningAndFailoverPath;
    public volatile boolean isShutdown = false;

    public FailoverListenerHolder(final TaskExecutor taskExecutor) {
        super(taskExecutor);
        configurationService = taskExecutor.getTaskConfigurationService();
        executeService = taskExecutor.getExecuteService();
        failoverService = taskExecutor.getFailoverService();
        executionPath = TaskNodePaths.getTaskNodePath(taskId, ExecuteNode.ROOT);
        runningAndFailoverPath = new HashSet<>();
    }

    @Override
    public void start() {
        nodeCacheOrTreeCacheManager.addTreeCacheListener(executionPath, 1, new ExecutionListener(this));
    }

    @Override
    public void stop() {
        super.stop();
        isShutdown = true;
        nodeCacheOrTreeCacheManager.closeTreeCache(executionPath, 1);
        closeRunningAndFailoverNodeCaches();
    }

    private void closeRunningAndFailoverNodeCaches() {
        Iterator<String> iterator = runningAndFailoverPath.iterator();
        while (iterator.hasNext()) {
            String next = iterator.next();
            nodeCacheOrTreeCacheManager.closeNodeCache(next);
        }
    }

    public synchronized void failover(final Integer taskItemIndex) {
        if (taskExecutor == null || taskExecutor.getTaskExecuteEngine() == null) {
            return;
        }
        if (!taskExecutor.getTaskExecuteEngine().isFailoverSupported() || !configurationService.isFailover() || executeService.isCompleted(taskItemIndex)) {
            return;
        }

        failoverService.createFailoverFlag(taskItemIndex);

        if (!executeService.hasRunningTaskItemList(taskExecutor.getShardingService().getShardingTaskItemIndexList())) {
            failoverService.failoverIfNecessary();
        }
    }

}
